package com.chinamobile.iot.kafka;

import com.alibaba.fastjson.JSON;
import com.chinamobile.iot.config.MqttConfiguration;
import com.chinamobile.iot.mqtt.spi.impl.ProtocolProcessor;
import com.chinamobile.iot.utils.NameThreadFactory;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by szl on 2017/7/10.
 */
@Component
public class KafkaDataProducer {

    private static final Logger logger = LoggerFactory.getLogger(KafkaDataProducer.class);

    @Autowired
    private KafkaProducerClient kafkaProducerClient;
    @Autowired
    private ProtocolProcessor protocolProcessor;
    private ExecutorService executorService;

    @Autowired
    private MqttConfiguration mqttConfiguration;

    @PostConstruct
    private void init() {
        if (mqttConfiguration.isKafka()) {
            executorService = Executors.newFixedThreadPool(1, new NameThreadFactory("DeviceDataAck"));
        }
    }

    @PreDestroy
    public void stop() {
        if (executorService != null) {
            executorService.shutdown();
        }
    }

    public void sendData(String deviceId, String topic, Object msg, int messageId) {

        ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, deviceId, JSON.toJSONString(msg));
        kafkaProducerClient.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception != null) {
                    logger.error("send data to kafka error! {}", exception.getMessage());
                } else {
                    executorService.execute(() -> protocolProcessor.sendPubAckByMqttId(deviceId, messageId));
                }
            }
        });
    }

    public void sendData(String deviceId, String topic, Object msg) {
        ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, deviceId, JSON.toJSONString(msg));
        kafkaProducerClient.send(record);
    }

}
